use crate::codec::d_codec::Packet;
use crate::config::config::Protocol;
use crate::config::Config;
use crate::serializer::serialization::Serialized;
use crate::start::dispatcher::{FnHandler, PacketDispatcher};
use crate::start::shutdown::Shutdown;
use crate::start::tcp::TcpServer;
use crate::start::udp::UdpServer;
use crate::start::{Connection, SendMsg};
use async_trait::async_trait;
use futures::future::Either;
use futures::FutureExt;
use std::borrow::Borrow;
use std::collections::HashMap;
use std::io;
use std::io::ErrorKind;
use std::net::{SocketAddr, ToSocketAddrs};
use std::sync::atomic::AtomicPtr;
use std::sync::Arc;
use tokio::net::TcpListener;
use tokio::runtime::Runtime;
use tokio::sync::mpsc;

pub enum State {
    Created,
    Initialized,
    Starting,
    Started,
    Shutdown,
}

pub struct Service {
    config: Config,
    state: AtomicPtr<State>,
    shutdown_handle: Shutdown,
    shutdown_receiver: Option<mpsc::Receiver<()>>,
    runtime: Option<Runtime>,
    dispatchers: Arc<PacketDispatcher>,
}

// #[derive(Default)]
pub struct ServiceBuilder {
    dispatchers: std::collections::HashMap<i16, Box<dyn crate::start::dispatcher::Handler>>,
    config: Config,
    state: AtomicPtr<State>,
}

impl ServiceBuilder {
    pub fn default(config: Config) -> Self {
        ServiceBuilder {
            dispatchers: Default::default(),
            config: config,
            state: Default::default(),
        }
    }
    pub fn register<F: 'static + Fn(Connection) + Sync + Send>(mut self, k: i16, f: F) -> Self {
        //     消息注册
        let mut x = &mut self.dispatchers;
        x.insert(k, Box::new(FnHandler::new(f)));
        self
    }
    pub fn builder(self) -> Service {
        let (shutdown_sender, shutdown_receiver) = mpsc::channel(1);
        let option = Runtime::new().ok();
        Service {
            config: self.config,
            state: self.state,
            shutdown_handle: Shutdown(shutdown_sender),
            shutdown_receiver: Some(shutdown_receiver),
            runtime: option,
            dispatchers: Arc::new(PacketDispatcher {
                dispatchers: self.dispatchers,
            }),
        }
    }
}

impl Service {
    // pub fn init() -> Service {
    //     let (shutdown_sender, shutdown_receiver) = mpsc::channel(1);
    //     let option = Runtime::new().ok();
    //     Service {
    //         config: Config::debug_default(),
    //         state: Default::default(),
    //         shutdown_handle: Shutdown(shutdown_sender),
    //         shutdown_receiver: Some(shutdown_receiver),
    //         runtime: option,
    //         dispatchers: Arc::new(PacketDispatcher::default()),
    //     }
    // }
    async fn accept_tcp(&self, listener: TcpListener) {
        let works = self.runtime.as_ref().unwrap();
        loop {
            let x = listener.accept().await.unwrap();
            works
                .spawn(async move { println!("{:?}", x.0) })
                .await
                .unwrap()
        }
    }
    pub async fn start(&mut self) {
        let full_addr = format!("{}:{}", self.config.address, self.config.port);
        let addr = full_addr
            .to_socket_addrs()
            .map(|mut addrs| addrs.next().expect(">= 1 socket addr"))
            .map_err(|e| ErrorKind::AddrNotAvailable)
            .unwrap();
        // If `ctrl-c` shutdown is enabled, we `select` on `the ctrl-c` signal
        // and server. Otherwise, we only wait on the `server`, hence `pending`.
        let shutdown_handle = self.shutdown_handle.clone();
        let shutdown_signal = match self.config.ctrl_c {
            true => tokio::signal::ctrl_c().boxed(),
            false => futures::future::pending().boxed(),
        };
        let dispatchers = self.dispatchers.clone();
        let arc = dispatchers;
        let server = {
            let runtime = self.runtime.as_ref().borrow().unwrap();
            match self.config.protocol {
                Protocol::UDP => {
                    let socket = UdpServer::bind(addr)
                        .await
                        .map_err(|e| ErrorKind::AddrNotAvailable)
                        .unwrap();
                    println!("start ok udp {:?}", addr);
                    UdpServer::poll(socket, arc).boxed()
                }
                _ => {
                    let l = TcpServer::bind( addr)
                        .await
                        .map_err(|e| ErrorKind::AddrNotAvailable)
                        .unwrap();
                    println!("start ok tcp {:?}", addr);
                    TcpServer::poll(l, arc).boxed()
                }
            }
        };

        match futures::future::select(shutdown_signal, server).await {
            Either::Left((Ok(()), server)) => {
                // Ctrl-was pressed. Signal shutdown, wait for the server.
                shutdown_handle.shutdown();
                server.await
            }
            Either::Left((Err(err), server)) => {
                // Error setting up ctrl-c signal. Let the user know.
                // warn!("Failed to enable `ctrl-c` graceful signal shutdown.");
                // info_!("Error: {}", err);
                server.await
            }
            // Server shut down before Ctrl-C; return the result.
            Either::Right((result, _)) => result,
        };
    }
    pub fn dispatcher(&self, con: Connection) {
        self.dispatchers.dispatcher(con);
    }

    #[inline(always)]
    pub fn shutdown(&self) -> Shutdown {
        self.shutdown_handle.clone()
    }
}

#[async_trait]
pub trait Server {
    type Socket;

    async fn bind(address: SocketAddr) -> io::Result<Self::Socket>;

    async fn poll(socket: Self::Socket, service: Arc<PacketDispatcher>);
}

pub struct Shared {
    pub peers: HashMap<SocketAddr, Tx>,
}

impl Shared {
    pub fn new() -> Self {
        Shared {
            peers: HashMap::new(),
        }
    }
}

/*----------------------------------Connection连接实现-----------------------------------------*/

/// Shorthand for the transmit half of the message channel.
pub type Tx = mpsc::UnboundedSender<bool>;

/// Shorthand for the receive half of the message channel.
pub type Rx = mpsc::UnboundedReceiver<bool>;

#[derive(Debug)]
pub enum Message {
    /// A message that should be broadcasted to others.
    Broadcast(Packet, SocketAddr),
    /// A message that should be Forward to others.
    Forward(Packet, SocketAddr),
    /// A message that should be received by a client
    Received(Packet, SocketAddr),

    Send(Packet, SocketAddr),

    Close(),
}
